﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.Util;
using System.Xml;

namespace Parallel3M.parallel3m.framework
{
    public class MQParallelSinker:ParallelProcessor
    {
        public string connectionURI="activemq:tcp://127.0.0.1:61616";
        public IConnectionFactory NMSfactory;
        public IConnection NMSconnection;
        public ISession NMSsession;
        public IDestination NMSdestination;
        public string NMSQueue;
        public IMessageProducer producer;
        
        
        public MQParallelSinker()
        {

        }

        public MQParallelSinker(string uri)
        {
            connectionURI = uri;
            try
            {
                NMSfactory = new NMSConnectionFactory(connectionURI);
                NMSconnection = NMSfactory.CreateConnection();
                NMSsession = NMSconnection.CreateSession();
                
                NMSdestination = SessionUtil.GetDestination(NMSsession, NMSQueue);
                producer = NMSsession.CreateProducer(NMSdestination);
                NMSconnection.Start();
            
            }
            catch
            {

            }
        }
             
        //send to the MQ services
        public override void process()
        {
            foreach (ParallelConnection inputConnection in inputConnections)
            {
                ParallelPayload payload = inputConnection.Flowout();
                if (payload != null)
                {
                    payloadcount++;

                    IMessage message = TransferPayload2Message(payload);
                    if (message != null)
                    {                        
                        producer.Send(message);
                    }

                }
            }           
        }

        public override void destroy()
        {
            NMSconnection.Close();
        }

        public ITextMessage TransferPayload2Message(ParallelPayload payload)
        {
            ITextMessage message = NMSsession.CreateTextMessage();
            if (payload.type == PAYLOADCONSTRANT.TYPE_REQUEST)
            {
                message.Text = payload.request;
            }
            else
            {
                message.Text = payload.response;            
            }

            payload.pcs_source = application.pcsid;
            message.Properties.SetString(ParallelPayload.ID, payload.id);
            message.Properties.SetString(ParallelPayload.ROUTETAG, payload.routeTag);
            message.Properties.SetString(ParallelPayload.ROUTETAG, payload.routeTag);
            message.Properties.SetString(ParallelPayload.TYPE, payload.type);
            message.Properties.SetString(ParallelPayload.ACCESS, payload.access);
            message.Properties.SetString(ParallelPayload.DOMAIN, payload.domain);
            message.Properties.SetString(ParallelPayload.HANDLEID, payload.handlerid);
            message.Properties.SetString(ParallelPayload.LANGUAGE, payload.language);            
            message.Properties.SetString(ParallelPayload.PCSSOURCE, payload.pcs_source);
            message.Properties.SetString(ParallelPayload.PCSDESTINATION, payload.pcs_des);
            
            message.Properties.SetInt(ParallelPayload.ROUTECOUNT, payload.routeCount); 
            return message;
        }


        //
        public override void PreParse()
        {
            base.PreParse();
            foreach (XmlNode childnode in node.ChildNodes)
            {
                if (childnode.Name == "url")
                {
                    connectionURI = childnode.InnerText;
                }
                else if (childnode.Name == "queneName")
                {
                    NMSQueue = childnode.InnerText;
                }
            }
            try
            {
                Uri connect_uri = new Uri(connectionURI);
                NMSfactory = new NMSConnectionFactory(connect_uri);
                NMSconnection = NMSfactory.CreateConnection();
                NMSsession = NMSconnection.CreateSession();
                
                NMSdestination = SessionUtil.GetDestination(NMSsession, NMSQueue);
                producer = NMSsession.CreateProducer(NMSdestination);
                NMSconnection.Start();
                
            }
            catch(Exception e)
            {
                Console.WriteLine(e.ToString());
            }           
        }

    }
}
